package com.ftf.tp.blog.infrastructure.util;


import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregation;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.cat.aliases.AliasesRecord;
import co.elastic.clients.elasticsearch.cat.indices.IndicesRecord;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.SourceConfig;
import co.elastic.clients.elasticsearch.indices.Alias;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.util.ObjectBuilder;
import com.alibaba.fastjson.JSONObject;
import com.ftf.tp.blog.infrastructure.config.EsQueryWrapper;
import com.ftf.tp.common.annotation.DocId;
import com.ftf.tp.common.annotation.EsClass;
import com.ftf.tp.common.annotation.EsField;
import com.ftf.tp.common.constants.EsDataType;
import com.ftf.tp.common.core.constants.BasePage;
import com.ftf.tp.common.core.exception.BusinessException;
import com.ftf.tp.common.core.utils.PageResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * es工具类
 *
 * @author tanghao
 */
@Component
public class ElasticsearchUtil {
    private static final Logger log = LoggerFactory.getLogger(ElasticsearchUtil.class);

    @Resource
    private ElasticsearchClient client;

    @Resource
    private ElasticsearchAsyncClient asyncClient;

    public ElasticsearchClient getClient() {
        return client;
    }

    public void setClient(ElasticsearchClient client) {
        this.client = client;
    }

    public ElasticsearchAsyncClient getAsyncClient() {
        return asyncClient;
    }

    public void setAsyncClient(ElasticsearchAsyncClient asyncClient) {
        this.asyncClient = asyncClient;
    }

    /**
     * 如果索引或者别名已经存在,那么就不会在创建了
     *
     * @return 是否成功
     */
    public <T> boolean createIndexSettingsMappings(Class<T> tClass) throws Exception {

        EsClass annotation = tClass.getAnnotation(EsClass.class);
        String index = getClassIndex(tClass);
        String alias = getClassAlias(tClass);
        List<String> indexs = indexs();
        List<String> aliases = aliases();
        if (indexs.contains(index) || aliases.contains(alias)) {
            return false;
        }
        int shards = annotation.shards();
        int replicas = annotation.replicas();
        StringBuilder stringBuilder = new StringBuilder("{");
        stringBuilder
                .append("\"settings\": {\n" + "    \"number_of_shards\": ")
                .append(shards).append(",\n").append("    \"number_of_replicas\": ")
                .append(replicas).append("\n").append("  },")
                .append("\"mappings\": {\n" + "    \"properties\": ");
        JSONObject jsonObject = new JSONObject();
        for (Field declaredField : tClass.getDeclaredFields()) {
            declaredField.setAccessible(true);
            JSONObject jsonObject1 = new JSONObject();
//            DocId DocId = declaredField.getAnnotation(DocId.class);
//            if (DocId != null) {
//                jsonObject1.put("type", EsDataType.LONG.getType());
//                jsonObject.put("id", jsonObject1);
//                continue;
//            }
            EsField annotation1 = declaredField.getAnnotation(EsField.class);
            if (annotation1 != null) {
                String name = annotation1.name();
                name = "".equals(name) ? declaredField.getName() : name;
                EsDataType type = annotation1.type();
                String analyzer = annotation1.analyzer();
                String searchAnalyzer = annotation1.searchAnalyzer();
                jsonObject1.put("type", type.getType());
                if (!"".equals(analyzer)) {
                    jsonObject1.put("analyzer", analyzer);
                }
                if (!"".equals(searchAnalyzer)) {
                    jsonObject1.put("search_analyzer", searchAnalyzer);
                }
                jsonObject.put(name, jsonObject1);
            }
        }
        Assert.isTrue(jsonObject.size() > 0, "请添加es相关注解");
        stringBuilder.append(jsonObject);
        stringBuilder.append("}}");
        Reader queryJson = new StringReader(stringBuilder.toString());
        CreateIndexRequest req = CreateIndexRequest.of(b -> b.index(index)
                .aliases(alias, Alias.of(a -> a.isWriteIndex(true)))
                .withJson(queryJson));

        return client.indices().create(req).acknowledged();
    }

    /**
     * 查询全部索引
     */
    public List<String> indexs() {
        List<IndicesRecord> indices = null;
        try {
            indices = client.cat().indices().valueBody();

        } catch (IOException e) {
            e.printStackTrace();
        }
        assert indices != null;
        return indices.stream().map(IndicesRecord::index).collect(Collectors.toList());
    }

    /**
     * 查询全部别名
     */
    public List<String> aliases() {
        List<AliasesRecord> aliasesRecords = null;
        try {
            aliasesRecords = client.cat().aliases().valueBody();
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert aliasesRecords != null;
        return aliasesRecords.stream().map(AliasesRecord::alias).collect(Collectors.toList());
    }

    /**
     * 查询索引下全部数据
     */
    public <T> List<T> queryAll(Class<T> tClass) {
        List<T> list = new ArrayList<>();
        SearchResponse<T> search = null;
        try {
            search = client.search(s ->
                    s.index(tClass.getAnnotation(EsClass.class).index())
                            .query(q -> q.matchAll(m -> m)), tClass);
            for (Hit<T> hit : search.hits().hits()) {
                setId(tClass, hit.source(), hit.id());
                list.add(hit.source());
            }
            return list;
        } catch (IOException e) {
            log.error("查询失败", e);
            throw new BusinessException("查询失败" + e.getMessage());
        }

    }

    /**
     * 分页条件查询
     */
    public <T> PageResult<T> page(Class<T> tClass, BasePage pageParams, EsQueryWrapper esQueryWrapper) {
        List<T> list = new ArrayList<>();
        SearchRequest.Builder builder = new SearchRequest.Builder();
        //todo 深度分页
        builder.from((pageParams.getPage() - 1) * pageParams.getSize());
        BoolQuery.Builder boolBuilder = esQueryWrapper.getBoolBuilder();
        SourceConfig.Builder sourceConfigBuilders = esQueryWrapper.getSourceConfigBuilders();
        SearchRequest searchRequest = builder
                .index(tClass.getAnnotation(EsClass.class).index())
                .query(boolBuilder.build()._toQuery())
                .source(sourceConfigBuilders.build())
                .size(pageParams.getSize())
                .trackTotalHits(t -> t.enabled(true))
                .build();

        SearchResponse<T> search = null;
        try {
            search = client.search(searchRequest, tClass);
            for (Hit<T> hit : search.hits().hits()) {
                setId(tClass, hit.source(), hit.id());
                list.add(hit.source());
            }
            return new PageResult<>(list, (int) search.hits().total().value(), pageParams.getPage(),
                    pageParams.getSize());
        } catch (Exception e) {
            log.error("分页查询失败", e);
            throw new BusinessException("分页查询失败" + e.getMessage());
        }
    }


    private <T> void setId(Class<T> tClass, Object data, String id) {
        try {
            Field[] fields = tClass.getDeclaredFields();
            for (Field field : fields) {
                if (field.getDeclaredAnnotation(DocId.class) != null) {
                    field.setAccessible(true);
                    field.set(data, Long.valueOf(id));
                }
            }
        } catch (IllegalAccessException e) {
            log.error("id设置失败", e);
            throw new BusinessException("id设置失败" + e.getMessage());
        }
    }


    /**
     * 添加数据
     *
     * @param data  数据
     * @param async 是否同步
     */
    public <T> String addData(T data, boolean async) {
        IndexResponse response = null;

        try {
            IndexRequest.Builder<T> indexReqBuilder = new IndexRequest.Builder<>();
            //取别名查询
            indexReqBuilder.index(getClassIndex(data.getClass()));
            indexReqBuilder.document(data);
            if (async) {
                asyncClient.index(indexReqBuilder.build());
                return null;
            }
            response = client.index(indexReqBuilder.build());
        } catch (IOException e) {
            log.error("增加数据失败", e);
        }
        assert response != null;
        return response.id();
    }

    /**
     * 更新数据
     *
     * @param data 数据
     */
    public <T> String update(T data) throws IOException {
        UpdateResponse<?> updateResponse = null;
        updateResponse = client.update(u -> u
                .index(getClassAlias(data.getClass()))
                .id((String) getId(data))
                .doc(data), data.getClass());

        assert updateResponse != null;
        return updateResponse.id();
    }

    /**
     * 批量添加数据
     *
     * @param list  数据
     * @param async 是否同步
     */
    public <T> void addDataList(List<T> list, boolean async) {
        BulkRequest.Builder br = new BulkRequest.Builder();
        for (T o : list) {
            br.operations(op -> op.index(idx -> idx.index(getClassIndex(o.getClass())).document(o)));
        }

        if (async) {
            asyncClient.bulk(br.build());
            return;
        }
        BulkResponse result = null;
        try {
            result = client.bulk(br.build());
        } catch (Exception e) {
            log.error("批量增加数据失败", e);
        }
        // Log errors, if any
        assert result != null;
        if (result.errors()) {
            log.error("Bulk had errors");
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
        }
    }

    /**
     * 批量更新数据
     *
     * @param list  数据
     * @param async 是否同步
     */
    public <T> void updateDataList(List<T> list, boolean async) {
        BulkRequest.Builder br = new BulkRequest.Builder();
        for (T o : list) {
            br.operations(op -> op.update(u -> u.index(getClassIndex(o.getClass()))
                    .id(String.valueOf(getId(o)))
                    .action(a -> a.doc(o))));

        }

        if (async) {
            asyncClient.bulk(br.build());
            return;
        }
        BulkResponse result = null;
        try {
            result = client.bulk(br.build());
        } catch (Exception e) {
            log.error("批量增加数据失败", e);
        }
        // Log errors, if any
        assert result != null;
        if (result.errors()) {
            log.error("Bulk had errors");
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
        }
    }

    /**
     * byId
     *
     * @param docId 文档id
     * @param clazz 数据类型
     */
    public <T> T byId(String docId, Class<T> clazz) {
        GetResponse<T> response;
        try {
            response = client.get(g -> g.index(getClassIndex(clazz)).id(docId), clazz);
            setId(clazz, response.source(), response.id());
            return response.source();
        } catch (IOException e) {
            log.error("查询失败", e);
            throw new BusinessException("查询失败" + e.getMessage());
        }
    }

    /**
     * deleteById
     *
     * @param docId 文档id
     * @param clazz 数据类型
     */
    public String deleteById(String docId, Class<?> clazz) {
        DeleteResponse response = null;
        try {
            response = client.delete(g -> g.index(getClassIndex(clazz)).id(docId));
            return response.result().jsonValue();
        } catch (IOException e) {
            log.error("删除文档失败", e);
            throw new BusinessException("删除文档失败" + e.getMessage());
        }
    }


    public <T> List<T> complexQuery(Query query, Class<T> clazz) throws IOException {
        List<T> list = new ArrayList<>();
        SearchResponse<T> response =
                client.search(s -> s.index(getClassIndex(clazz)).query(query), clazz);
        List<Hit<T>> hits = response.hits().hits();
        for (Hit<T> hit : hits) {
            list.add(hit.source());
        }
        return list;
    }

    public <T> Aggregate complexQueryAggregations(
            Query query, Function<Aggregation.Builder, ObjectBuilder<Aggregation>> fn, Class<T> clazz)
            throws IOException {

        SearchResponse<T> response =
                client.search(
                        s ->
                                s.index(getClassIndex(clazz))
                                        .size(0) // 不需要显示数据 ,只想要聚合结果
                                        .query(query)
                                        .aggregations("aggregations", fn),
                        clazz);

        return response.aggregations().get("aggregations");
    }


    /**
     * 获取类的索引名称(没有指定默认类名首字母小写,  前缀+索引)
     */
    private <T> String getClassIndex(Class<T> clazz) {
        EsClass annotation = clazz.getAnnotation(EsClass.class);
        String index = annotation.index();
        index = "".equals(index)
                ? Objects.requireNonNull(clazz.getName()).toLowerCase()
                : index.toLowerCase();
        return index;
    }

    /**
     * 获取类的索引名称(没有指定默认类名首字母小写,  前缀+索引)
     */
    private Object getId(Object o) {
        Class<?> clazz = o.getClass();
        Field[] declaredFields = clazz.getDeclaredFields();
        try {
            for (Field field : declaredFields) {
                if (field.getDeclaredAnnotation(DocId.class) != null) {
                    field.setAccessible(true);
                    return field.get(o);
                }
            }
        } catch (IllegalAccessException e) {
            log.error("反射获取对象id的值失败", e);
        }
        return null;
    }

    private <T> String getClassAlias(Class<T> clazz) {
        EsClass annotation = clazz.getAnnotation(EsClass.class);
        String alias = annotation.alias();
        alias =
                "".equals(alias)
                        ? Objects.requireNonNull(clazz.getName()).toLowerCase()
                        : alias.toLowerCase();
        return alias;
    }


}


